package kafka_yl.producer;

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

public class KafkaProducerTest implements Runnable{
	private final KafkaProducer<String,String> producer;
	private final String topic;
	public KafkaProducerTest(String topicName) {
        Properties props = new Properties();
        //bootstrap.servers： kafka的地址
        props.put("bootstrap.servers", "192.168.19.161:6667");
        //acks:消息的确认机制，默认值是0
        //0,生产者不会等待kafka的响应
        //1,kafka会把这条消息写到本地日志文件中，但是不会等待集群中其他机器的成功响应
        //all,leader会等待所有的follower同步完成,确保消息不会丢失，除非kafka集群中所有机器挂掉
        props.put("acks", "all");
        //retries：配置为大于0的值的话，客户端会在消息发送失败时重新发送
        props.put("retries", 0);
        //batch.size:当多条消息需要发送到同一个分区时，生产者会尝试合并网络请求,提高client和生产者的效率
        //16384是默认值，单位bytes,即消息达到16kB就合并发送
        props.put("batch.size", 16384);
        //键序列化
        props.put("key.serializer", StringSerializer.class.getName());
        //值序列化
        props.put("value.serializer", StringSerializer.class.getName());
        this.producer = new KafkaProducer<String, String>(props);
        this.topic = topicName;
    }
	@Override
	public void run() {
		int messageNo=1;
		try {
			for(;;) {
				String messageStr="这是第"+messageNo+"条数据";
				//producer.send(new ProducerRecord<String, String>(topic,key,value));
				//开始生产，topic是消息队列名称，未创建会自动创建；key是键值，value是发送的数据，与key成对
				producer.send(new ProducerRecord<String, String>(topic, "Message",messageStr));
				if(messageNo%10==0) {
					System.out.println("发送的消息："+messageStr);
				}
				if(messageNo%100==0) {
					System.out.println("成功发送了"+messageNo+"条数据");
					break;
				}
				messageNo++;
			}
		} catch (Exception e) {
			e.printStackTrace();
		}finally {
			producer.close();
		}
	}
	public static void main(String args[]) {
		KafkaProducerTest test=new KafkaProducerTest("KAFKA_YL");
		Thread thread=new Thread(test);
		thread.start();
	}

}
